// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: linchuan

#include "coin2/exchange/huobi/order_v1/parser.h"

#include <memory>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include "coin/proto/coin_request.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/huobi/api_order/native_private_client.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::huobi::order_v1 {

namespace {

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferBundle;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::huobi::symbology::HuobiCurrency;
using coin2::exchange::huobi::symbology::HuobiProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using HuobiOrderSubTopicId = coin2::exchange::huobi::api_order::HuobiOrderSubTopicId;

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("client-order-id")) {
    if (data["client-order-id"].IsString()) {
      const std::string& idstr = data["client-order-id"].GetString();
      if (idstr.length() > 0) {
        proc_order_id = stoll(idstr);
      }
    } else if (data["client-order-id"].IsInt64()) {
      proc_order_id = std::stoll(data["client-order-id"].GetString());
    }
  } else if (data.HasMember("clientOrderId") && data["clientOrderId"].IsString()) {
    proc_order_id = stoll(data["clientOrderId"].GetString());
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& order_type) {
  // buy-market, sell-market, buy-limit, sell-limit, buy-ioc, sell-ioc,
  // buy-limit-maker, sell-limit-maker,buy-stop-limit,sell-stop-limit
  if (order_type.find("buy") != std::string::npos) {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (order_type.find("sell") != std::string::npos) {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& order_type) {
  // buy-market, sell-market, buy-limit, sell-limit, buy-ioc, sell-ioc,
  // buy-limit-maker, sell-limit-maker,buy-stop-limit,sell-stop-limit
  if (order_type.find("ioc") != std::string::npos) {
    return coin::proto::OrderType::IOC_LIMIT_ORDER;
  } else if (order_type.find("market") != std::string::npos) {
    return coin::proto::OrderType::MARKET_ORDER;
  } else if (order_type.find("limit") != std::string::npos) {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline coin::proto::FillType GetFillType(const std::string_view& fill_type) {
  if (fill_type == "maker") {
    return FillType::MAKER_FILL_TYPE;
  } else if (fill_type == "taker") {
    return FillType::TAKER_FILL_TYPE;
  } else {
    return FillType::UNKNOWN_FILL_TYPE;
  }
}

inline void SetOrderBasicInfo(ProductOrderElement* order) {
  order->set_market_type(MarketType::Spot);
  order->set_exchange_type(ExchangeType::Huobi);
  order->set_api_version("v1");
  return;
}

inline int64_t GetTimestampFromWs2(const JsonValue& data) {
  int64_t ts;
  if (data.HasMember("orderCreateTime") && data["orderCreateTime"].IsInt64()) {
    ts = std::stoll(data["orderCreateTime"].GetString()) * 1000000LL;
  } else if (data.HasMember("tradeTime") && data["tradeTime"].IsInt64()) {
    ts = std::stoll(data["tradeTime"].GetString()) * 1000000LL;
  } else if (data.HasMember("lastActTime") && data["lastActTime"].IsInt64()) {
    ts = std::stoll(data["lastActTime"].GetString()) * 1000000LL;
  } else {
    ts = GetCurrentTimestamp();
  }
  return ts;
}

inline void SetOrderStatusFromWs(const std::string& status, ProductOrderElement* order) {
  if (status == "submitted" || status == "partial-filled") {
    order->set_is_live(true);
  } else if (status == "canceled" || status == "partial-canceled") {
    order->set_cancel_confirmed(true);
  } else if (status == "filled") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseWsOrderInfo] Unknown order status! " << status;
  }
}

inline std::string GetExchangeOrderErrorCode(const JsonValue& json) {
  if (json.HasMember("err-code") && json["err-code"].IsString()) {
    return json["err-code"].GetString();
  } else {
    return {};
  }
}

inline std::string GetExchangeOrderErrorMsg(const JsonValue& json) {
  if (json.HasMember("err-msg") && json["err-msg"].IsString()) {
    return json["err-msg"].GetString();
  } else {
    return {};
  }
}

inline coin::proto::OrderErrorCode GetOrderErrorCode(const JsonValue& json_error_code) {
  using coin::proto::OrderErrorCode;
  auto result = OrderErrorCode::ORDER_UNKNOWN_ERROR;
  std::string_view error_code = json_error_code.GetString();

  if (error_code == "order-accountbalance-error" ||
      error_code == "account-frozen-balance-insufficient-error") {
    result = OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND;
  } else if (error_code == "order-orderstate-error") {
    result = OrderErrorCode::ORDER_NOT_FOUND;
  } else if (error_code == "order-limitorder-price-error") {
    result = OrderErrorCode::SUBMIT_ORDER_WRONG_PRICE;
  } else if (error_code == "order-limitorder-amount-error") {
    result = OrderErrorCode::SUBMIT_ORDER_WRONG_QTY;
  } else if (error_code == "order-orderprice-precision-error") {
    result = OrderErrorCode::SUBMIT_ORDER_WRONG_PRICE;
  } else if (error_code == "order-orderamount-precision-error") {
    result = OrderErrorCode::SUBMIT_ORDER_WRONG_QTY;
  }

  return result;
}  // namespace

bool IsHuobiErrMessage(const JsonValue& doc) {
  return doc.HasMember("status") && strcmp(doc["status"].GetString(), "error") == 0;
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!(json.HasMember("ch") && json.HasMember("action"))) {
    return false;
  }

  return (
      (std::strncmp(json["ch"].GetString(), "orders", 6) == 0 ||
       std::strncmp(json["ch"].GetString(), "trade", 5) == 0) &&
      std::strncmp(json["action"].GetString(), "push", 4) == 0);
}

}  // namespace

void HuobiParser::MergeMarginAccountBalance(AccountBalance* acc_balance) {
  acc_balance->Clear();
  acc_balance->set_exchange("Huobi");
  acc_balance->set_market_type("Margin");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeLower()) {
    auto* balance = acc_balance->add_each_balance();
    balance->set_total(0);
    balance->set_hold(0);
    balance->set_available(0);
    if (isolated_margin_balance_map_cache_.count(native_currency)) {
      auto& tmp = isolated_margin_balance_map_cache_.at(native_currency);
      balance->set_total(balance->total() + tmp.total());
      balance->set_hold(balance->hold() + tmp.hold());
      balance->set_available(balance->available() + tmp.available());
    }
    if (crossed_margin_balance_map_cache_.count(native_currency)) {
      auto& tmp = crossed_margin_balance_map_cache_.at(native_currency);
      balance->set_total(balance->total() + tmp.total());
      balance->set_hold(balance->hold() + tmp.hold());
      balance->set_available(balance->available() + tmp.available());
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
}

void HuobiParser::ParseRestSpotAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestSpotAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

void HuobiParser::ParseRestMarginCrossedAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestMarginCrossedAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  RawBalanceMap raw_balance_map;
  auto success = ParseRawMarginBalance(document, &raw_balance_map);
  if (!success) {
    return;
  }
  crossed_margin_balance_map_cache_ = std::move(raw_balance_map);
  is_crossed_margin_balance_initialized = true;
  if (!is_isolated_margin_balance_initialized) {
    return;
  }

  AccountBalance acc_balance;
  MergeMarginAccountBalance(&acc_balance);
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(acc_balance);
}

void HuobiParser::ParseRestMarginIsolatedAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestMarginIsolatedAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  RawBalanceMap raw_balance_map;
  auto success = ParseRawMarginBalance(document, &raw_balance_map);
  if (!success) {
    return;
  }
  isolated_margin_balance_map_cache_ = std::move(raw_balance_map);
  is_isolated_margin_balance_initialized = true;
  if (!is_crossed_margin_balance_initialized) {
    return;
  }

  AccountBalance acc_balance;
  MergeMarginAccountBalance(&acc_balance);

  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(acc_balance);
}

bool HuobiParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "data": {
      "id": 100009,
      "type": "spot",
      "state": "working",
      "list": [
        {
          "currency": "usdt",
          "type": "trade",
          "balance": "5007.4362872650"
        },
        {
          "currency": "usdt",
          "type": "frozen",
          "balance": "348.1199920000"
        }
      ]
    }
  }
  */
  if (doc.IsObject() && IsHuobiErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("data") || !doc["data"].HasMember("list") || !doc["data"]["list"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  bool current_trade_init = false;
  double current_trade_balance = 0;
  std::string_view current_trade_currency("");
  for (const auto& data : doc["data"]["list"].GetArray()) {
    if (!data.HasMember("currency") || !data.HasMember("balance")) {
      continue;
    }
    std::string_view type_str = data["type"].GetString();
    if (type_str == "trade") {
      current_trade_init = true;
      current_trade_balance = stod(data["balance"].GetString());
      current_trade_currency = data["currency"].GetString();
    } else if (type_str == "frozen") {
      if (!current_trade_init || data["currency"].GetString() != current_trade_currency) {
        continue;
      }
      double current_frozen_balance = stod(data["balance"].GetString());
      std::string current_trade_currency_str(current_trade_currency);
      CurrencyBalance balance;
      balance.set_currency_native(current_trade_currency_str);
      balance.set_available(current_trade_balance);
      balance.set_hold(current_frozen_balance);
      balance.set_total(current_trade_balance + current_frozen_balance);
      raw_balance_map->emplace(balance.currency_native(), balance);
    }
  }

  return true;
}

bool HuobiParser::ParseRawMarginBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "data": {
      "id": 100009,
      "type": "spot",
      "state": "working",
      "list": [
        {
          "currency": "usdt",
          "type": "trade",
          "balance": "5007.4362872650"
        },
        {
          "currency": "usdt",
          "type": "frozen",
          "balance": "348.1199920000"
        },
        {
          "currency": "usdt",
          "type": "loan",
          "balance": "100.1199920000"
        }
      ]
    }
  }
  */
  if (doc.IsObject() && IsHuobiErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("data") || !doc["data"].HasMember("list") || !doc["data"]["list"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  std::string_view current_trade_currency("");
  for (const auto& data : doc["data"]["list"].GetArray()) {
    if (!data.HasMember("currency") || !data.HasMember("balance")) {
      continue;
    }
    std::string_view type_str = data["type"].GetString();
    current_trade_currency = data["currency"].GetString();
    std::string current_trade_currency_str = std::string(current_trade_currency);
    auto& balance = (*raw_balance_map)[current_trade_currency_str];
    if (!balance.has_currency_native()) {
      balance.set_currency_native(current_trade_currency_str);
    }
    if (type_str == "trade" || type_str == "loan" || type_str == "interest") {
      balance.set_available(balance.available() + stod(data["balance"].GetString()));
    } else if (type_str == "frozen") {
      balance.set_hold(balance.hold() + stod(data["balance"].GetString()));
    }
  }
  for (auto& [currency, balance] : *raw_balance_map) {
    balance.set_total(balance.available() + balance.hold());
  }

  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance HuobiParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  bool success = false;
  success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Huobi");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeLower()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

ProductOrderElement HuobiParser::ParseRestOrderInfo(const JsonValue& data) {
  ProductOrderElement order;
  SetOrderBasicInfo(&order);
  order.set_exchange_order_id(data["id"].GetString());
  order.set_symbol(data["symbol"].GetString());
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["amount"].GetString()));
  order.set_proc_order_id(GetProcOrderId(data));
  order.set_side(GetOrderSide(data["type"].GetString()));
  order.set_order_type(GetOrderType(data["type"].GetString()));
  order.set_timestamp(msg_->timestamp);
  const std::string_view status = data["state"].GetString();
  if (status == "created" || status == "cancelling" || status == "pre-submitted") {
    order.set_is_live(false);
  } else if (status == "submitted" || status == "partial-filled") {
    order.set_is_live(true);
  } else if (status == "canceled" || status == "partial-canceled") {
    order.set_cancel_confirmed(true);
  } else if (status == "filled") {
    order.set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
  return order;
}

void HuobiParser::ParseRestOrdersInfo() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOrdersInfo] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsHuobiErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] " << JsonToString(data);
    return;
  }

  if (!data.HasMember("data")) {
    LOG(ERROR) << "Not .data in json. " << JsonToString(data);
    return;
  }

  const auto& info_list = data["data"];
  if (!info_list.IsArray()) {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& info : info_list.GetArray()) {
    auto* order = product_order_bundle->add_each_order();
    auto new_order = ParseRestOrderInfo(info);
    order->MergeFrom(new_order);
    product_order_bundle_found = true;
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

bool HuobiParser::ParseWsOrderInfo(const JsonValue& data, ProductOrderElement* order) {
  /*
  {
    "action":"push",
    "ch":"orders#btcusdt",
    "data":
    {
        "orderSize":"2.000000000000000000",
        "orderCreateTime":1583853365586,
        "orderPrice":"77.000000000000000000",
        "type":"sell-limit",
        "orderId":27163533,
        "clientOrderId":"liujin",
        "orderStatus":"submitted",
        "symbol":"btcusdt",
        "eventType":"creation"
    }
  }
  {
    "action":"push",
    "ch":"orders#btcusdt",
    "data":
    {
        "tradePrice":"76.000000000000000000",
        "tradeVolume":"1.013157894736842100",
        "tradeId":301,
        "tradeTime":1583854188883,
        "aggressor":true,
        "remainAmt":"0.000000000000000400000000000000000000",
        "orderId":27163536,
        "clientOrderId":"",
        "orderStatus":"filled",
        "symbol":"btcusdt",
        "eventType":"trade"
    }
  }
  {
    "action":"push",
    "ch":"orders#btcusdt",
    "data":
    {
        "lastActTime":1583853475406,
        "remainAmt":"2.000000000000000000",
        "orderId":27163533,
        "clientOrderId":"liujin",
        "orderStatus":"canceled",
        "symbol":"btcusdt",
        "eventType":"cancellation"
    }
  }
  */
  SetOrderBasicInfo(order);
  order->set_timestamp(msg_->timestamp);
  order->set_exchange_order_id(data["orderId"].GetString());
  order->set_proc_order_id(GetProcOrderId(data));
  order->set_symbol(data["symbol"].GetString());
  SetOrderStatusFromWs(data["orderStatus"].GetString(), order);
  if (std::string_view(data["eventType"].GetString()) == "creation") {
    order->set_price(stod(data["orderPrice"].GetString()));
    order->set_qty(stod(data["orderSize"].GetString()));
    order->set_side(GetOrderSide(data["type"].GetString()));
    order->set_order_type(GetOrderType(data["type"].GetString()));
  } else {
    auto* oc = ocm_->FindByExchangeOrderId(order->exchange_order_id());
    if (!oc) {
      oc = ocm_->FindByProcOrderId(order->proc_order_id());
    }
    if (!oc) {
      return false;
    }
  }

  return true;
}

ProductOrderBundle HuobiParser::ParseWsOrderInfoListV2(const JsonValue& data) {
  ProductOrderBundle product_order_bundle;
  if (!data.HasMember("data")) {
    LOG(ERROR) << "Not .data in json. " << JsonToString(data);
    return product_order_bundle;
  }

  ProductOrderElement new_order;
  if (ParseWsOrderInfo(data["data"], &new_order)) {
    auto* order = product_order_bundle.add_each_order();
    order->MergeFrom(new_order);
  }
  return product_order_bundle;
}

ProductFillBundle HuobiParser::ParseWsFillInfoListV2(const JsonValue& data) {
  /*
  {
    "ch": "trade.clearing#btcusdt",
    "data": {
         "symbol": "btcusdt",
         "orderId": 99998888,
         "tradePrice": "9999.99",
         "tradeVolume": "0.96",
         "orderSide": "buy",
         "aggressor": true,
         "tradeId": 919219323232,
         "tradeTime": 998787897878,
         "transactFee": "19.88",
         "feeDeduct": "0",
         "feeDeductType": ""
    }
  }
  */
  ProductFillBundle product_fill_bundle;
  if (!data.HasMember("data")) {
    return product_fill_bundle;
  }

  const auto& info = data["data"];
  auto* fill = product_fill_bundle.add_each_fill();
  fill->set_symbol(info["symbol"].GetString());
  fill->set_fill_timestamp(msg_->timestamp);
  fill->set_exchange_order_id(info["orderId"].GetString());
  fill->set_price(stod(info["tradePrice"].GetString()));
  fill->set_qty(stod(info["tradeVolume"].GetString()));
  fill->set_fill_id(info["tradeId"].GetString());
  if (info["aggressor"].GetBool()) {
    fill->set_fill_type(FillType::TAKER_FILL_TYPE);
  } else {
    fill->set_fill_type(FillType::MAKER_FILL_TYPE);
  }
  auto native_currency = info["feeCurrency"].GetString();
  if (symcache_->has_currency_native(native_currency)) {
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    fill->set_fee_currency(currency->currency());
    fill->set_fee(stod(info["transactFee"].GetString()));
  }

  return product_fill_bundle;
}

void HuobiParser::ParseRestFillInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // response see v1/order/matchresults
  if (IsHuobiErrMessage(data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] " << JsonToString(data);
    return;
  }
  if (!(data.HasMember("data") && data["data"].IsArray() && data["data"].Size() > 0)) {
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (const auto& info : data["data"].GetArray()) {
    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(info["symbol"].GetString());
    fill->set_exchange_order_id(info["order-id"].GetString());
    fill->set_price(stod(info["price"].GetString()));
    fill->set_qty(stod(info["filled-amount"].GetString()));
    fill->set_fill_id(info["trade-id"].GetString());  // ws1 use match-id
    fill->set_fill_type(GetFillType(info["role"].GetString()));
    auto native_currency = info["fee-currency"].GetString();
    if (symcache_->has_currency_native(native_currency)) {
      auto* currency = symcache_->GetCurrencyByNative(native_currency);
      fill->set_fee_currency(currency->currency());
      fill->set_fee(stod(info["filled-fees"].GetString()));
    }
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void HuobiParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(static_cast<int32_t>(http_context->res.base().result()));
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "AsioHttpContextStatus Error When Submitting Order";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[HuobiSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[HuobiSubmitOrder] Parse Error\n" << http_context->res.body();
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }
  if (json.HasMember("err-code")) {
    auto error_code = GetOrderErrorCode(json["err-code"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[Huobi Submit Order] Error Response" << JsonToString(json);
    }
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  CHECK_THROW(json.HasMember("data")) << JsonToString(json);
  resp->set_exchange_order_id(json["data"].GetString());
}

void HuobiParser::ParseBatchSubmitResponse() {
  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_SUBMIT_RESPONSE);
  auto* response = result_.mutable_multi_submit_response();
  response->Clear();

  std::unordered_map<int64_t, SubmitOrderResponse*> proc_order_id_to_response_map;
  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_submit_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(static_cast<int32_t>(http_context->res.base().result()));
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);

    CHECK(proc_order_id_to_response_map.count(order_context->proc_order_id()) == 0)
        << "proc_order_id duplicate proc_order_id:" << order_context->proc_order_id();
    proc_order_id_to_response_map[order_context->proc_order_id()] = resp;
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json) || !json.IsObject() || !json.HasMember("data") ||
      !json["data"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  const auto& array = json["data"].GetArray();

  CHECK_EQ(proc_order_id_to_response_map.size(), array.Size())
      << "Batch submit response size not equal to request size response:" << JsonToString(json)
      << " request:" << response->DebugString();
  for (const auto& data : array) {
    int64_t client_order_id = GetProcOrderId(data);

    CHECK(proc_order_id_to_response_map.count(client_order_id) != 0)
        << "unknown client order id found client_order_id:" << client_order_id;

    auto* resp = proc_order_id_to_response_map[client_order_id];

    if (data.HasMember("err-code")) {
      auto error_code = GetOrderErrorCode(data["err-code"]);
      resp->set_error_code(error_code);
      resp->set_exchange_error_code(GetExchangeOrderErrorCode(data));
      resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(data));
      if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
        LOG(ERROR) << "[Huobi Submit Order] Error Response" << JsonToString(data);
      }
      continue;
    }

    resp->set_success(true);
    CHECK_THROW(data.HasMember("order-id")) << JsonToString(json);
    resp->set_exchange_order_id(data["order-id"].GetString());
  }
}

void HuobiParser::ParseCancelResponse() {
  /*
   * {
   *   "data": "59378"
   * }
   * {
   *   "status": "error",
   *   "err-code": "order-orderstate-error",
   *   "err-msg": "some message",
   *   "order-state":-1  // current order state
   * }
   * */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[HuobiCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[HuobiCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("err-code")) {
    auto error_code = GetOrderErrorCode(json["err-code"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      resp->set_success(false);
      return;
    }
  }

  if (json.HasMember("order_id")) {
    resp->set_exchange_order_id(std::string(json["order_id"].GetString()));
  }
  resp->set_success(true);
}

void HuobiParser::ParseBatchCancelResponse() {
  /*
  {
      "status":"ok",
      "data":{
          "success":[
              "829308976207525"
          ],
          "failed":[
              {
                  "err-msg":"Incorrect order state",
                  "order-state":6,
                  "order-id":"829309295200614",
                  "err-code":"order-orderstate-error",
                  "client-order-id":null
              }
          ]
      }
  }
  */

  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_CANCEL_RESPONSE);
  auto* response = result_.mutable_multi_cancel_response();
  response->Clear();

  std::unordered_map<int64_t, CancelOrderResponse*> proc_order_id_to_response_map;
  std::unordered_map<std::string, CancelOrderResponse*> order_id_to_response_map;
  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_cancel_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.result_int());
    resp->set_exchange_order_id(order_context->exchange_order_id());
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);

    CHECK(proc_order_id_to_response_map.count(order_context->proc_order_id()) == 0)
        << "proc_order_id duplicate proc_order_id:" << order_context->proc_order_id();
    proc_order_id_to_response_map[order_context->proc_order_id()] = resp;

    if (!order_context->has_exchange_order_id()) {
      continue;
    }
    CHECK(order_id_to_response_map.count(order_context->exchange_order_id()) == 0)
        << "exchange_order_id duplicate exchange_order_id:" << order_context->exchange_order_id();
    order_id_to_response_map[order_context->exchange_order_id()] = resp;
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json) || !json.IsObject() || !json.HasMember("data") ||
      !json["data"].IsObject() || !json["data"].HasMember("success") ||
      !json["data"].HasMember("failed") || !json["data"]["success"].IsArray() ||
      !json["data"]["failed"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  const auto& success = json["data"]["success"].GetArray();
  const auto& failed = json["data"]["failed"].GetArray();

  CHECK_EQ(proc_order_id_to_response_map.size(), success.Size() + failed.Size())
      << "Batch cancel response size not equal to request size response:" << JsonToString(json)
      << " request:" << response->DebugString();

  for (const auto& id : success) {
    std::string order_id = id.GetString();

    if (order_id_to_response_map.count(order_id) != 0) {
      auto* resp = order_id_to_response_map[order_id];
      resp->set_exchange_order_id(order_id);
      resp->set_success(true);

      continue;
    }

    int64_t proc_order_id = stoll(order_id);

    if (proc_order_id_to_response_map.count(proc_order_id) != 0) {
      auto* resp = proc_order_id_to_response_map[proc_order_id];
      resp->set_success(true);

      continue;
    }

    NOTREACHED() << "Not found id:" << order_id
                 << " in order_ids and proc_order_ids! response:" << JsonToString(json);
  }

  for (const auto& failed_resp : failed) {
    CancelOrderResponse* resp = nullptr;

    if (failed_resp.HasMember("order-id") &&
        order_id_to_response_map.count(failed_resp["order-id"].GetString()) != 0) {
      resp = order_id_to_response_map[failed_resp["order-id"].GetString()];
      resp->set_exchange_order_id(failed_resp["order-id"].GetString());
    } else {
      auto proc_order_id = GetProcOrderId(failed_resp);
      if (proc_order_id != 0 && proc_order_id_to_response_map.count(proc_order_id) != 0) {
        resp = proc_order_id_to_response_map[proc_order_id];
      }
    }

    CHECK_NE(resp, nullptr) << "Not found by order_id and proc_order_id, failed_response:"
                            << JsonToString(failed_resp);

    auto error_code = GetOrderErrorCode(failed_resp["err-code"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(failed_resp));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(failed_resp));
  }
}

void HuobiParser::ParseRestDeposit() {
  /*
  {
    "status":"ok",
    "data":[
      {
        "id": "8817904",
        "type": "deposit",
        "sub-type": "NORMAL",
        "currency": "btc",
        "chain": "btc",
        "tx-hash": "",
        "amount": "0.500000000000000000",
        "from-addr-tag": "",
        "address": "166drZkpjfrAcFfwoV2zCkF6UXYnDAEvGF",
        "address-tag": "",
        "fee": "0",
        "state": "safe",
        "wallet-confirm": "6",
        "created-at": "1544694256252",
        "updated-at": "1544695332868"
      }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestDeposit] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsHuobiErrMessage(document)) {
    LOG(ERROR) << "[ParseRestDeposit] Error message\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestDeposit] Invalid data format\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("currency") || !data.HasMember("amount") || !data.HasMember("type")) {
      LOG(ERROR) << "[ParseRestDeposit] Invalid transfer format";
      continue;
    }
    if (!data.HasMember("tx-hash") || data["tx-hash"].IsNull() ||
        (std::string(data["tx-hash"].GetString()) == "")) {
      LOG(INFO) << "[ParseRestDeposit] Ignore none on-chain tranfer";
      continue;
    }
    if (std::string(data["type"].GetString()) != "deposit") {
      continue;
    }

    std::string transfer_id = data["tx-hash"].GetString();
    if (transfer_deposit_map_.count(transfer_id) <= 0) {
      transfer_deposit_map_[transfer_id] = "";
    }
    std::string state = data["state"].GetString();
    if ((state != "confirming") && (state != "confirmed") && (state != "safe") &&
        (state != "orphan")) {
      continue;
    }
    if (!transfer_first_record_) {
      if (state == transfer_deposit_map_[transfer_id]) {
        continue;
      }
    }

    transfer_deposit_map_[transfer_id] = state;

    int64_t create_time = stoll(data["created-at"].GetString()) * 1e6;
    auto currency = HuobiCurrency::FromNativeStrImpl(data["currency"].GetString(), create_time);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::DEPOSIT);
    if (state == "confirming") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if ((state == "confirmed") || (state == "safe") || (state == "orphan")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_external_transfer_id(transfer_id);
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_from_address(data["address"].GetString());
    currency_transfer->set_create_timestamp(create_time);
    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!is_transfer_deposit_initialized_) {
      is_transfer_deposit_initialized_ = true;
      LOG(INFO) << "[ParseRestDeposit] Deposit ignored: " << transfer_deposit_map_.size();
      return;
    }
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestDeposit] Deposit parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

void HuobiParser::ParseRestWithdrawal() {
  /*
  {
    "status":"ok",
    "data":[
      {
        "id":45182894,
        "type":"withdraw",
        "sub-type":"FAST",
        "currency":"usdt",
        "chain":"trc20usdt",
        "tx-hash":"",
        "amount":400,
        "from-addr-tag":"",
        "address":"TRwkUYHWgUh23jbKpgTcYHgE9CcBzhGno9",
        "address-tag":"",
        "fee":0,
        "state":"confirmed",
        "created-at":1612261330443,
        "updated-at":1612261389250
      }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsHuobiErrMessage(document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestWithdrawal] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("currency") || !data.HasMember("amount") || !data.HasMember("type")) {
      LOG(ERROR) << "[ParseRestWithdrawal] Invalid transfer format";
      continue;
    }
    if (!data.HasMember("tx-hash") || data["tx-hash"].IsNull() ||
        (std::string(data["tx-hash"].GetString()) == "")) {
      LOG(INFO) << "[ParseRestWithdrawal] Ignore none on-chain tranfer";
      continue;
    }
    if (std::string(data["type"].GetString()) != "withdraw") {
      continue;
    }

    std::string transfer_id = data["tx-hash"].GetString();
    if (transfer_withdrawal_map_.count(transfer_id) <= 0) {
      transfer_withdrawal_map_[transfer_id] = "";
    }
    std::string state = data["state"].GetString();
    if ((state != "verifying") && (state != "failed") && (state != "submitted") &&
        (state != "reexamine") && (state != "canceled") && (state != "pass") &&
        (state != "reject") && (state != "pre-transfer") && (state != "wallet-transfer") &&
        (state != "wallet-reject") && (state != "confirmed") && (state != "confirm-error") &&
        (state != "repealed")) {
      continue;
    }
    if (!transfer_first_record_) {
      if (state == transfer_withdrawal_map_[transfer_id]) {
        continue;
      }
    }

    transfer_withdrawal_map_[transfer_id] = state;

    int64_t create_time = stoll(data["created-at"].GetString()) * 1e6;
    auto currency = HuobiCurrency::FromNativeStrImpl(data["currency"].GetString(), create_time);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::WITHDRAW);
    if (state == "canceled") {
      currency_transfer->set_status(TransferRequestProto::CANCEL_CONFIRMED);
    } else if (
        (state == "failed") || (state == "reject") || (state == "wallet-reject") ||
        (state == "confirm-error") || (state == "repealed")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_REJECTED);
    } else if (
        (state == "verifying") || (state == "reexamine") || (state == "submitted") ||
        (state == "pre-transfer") || (state == "wallet-transfer")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if (state == "pass") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_ACCEPTED);
    } else if (state == "confirmed") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_external_transfer_id(transfer_id);
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_to_address(data["address"].GetString());
    currency_transfer->set_create_timestamp(create_time);
    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!is_transfer_withdrawal_initialized_) {
      is_transfer_withdrawal_initialized_ = true;
      LOG(INFO) << "[ParseRestWithdrawal] Withdrawal ignored: " << transfer_withdrawal_map_.size();
      return;
    }
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestWithdrawal] Withdrawal parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

void HuobiParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsOrderUpdate(document)) {
    if (std::strncmp(document["ch"].GetString(), "orders", 6) == 0) {
      *result_.mutable_order_list() = ParseWsOrderInfoListV2(document);
      result_.set_type(MsgType::ORDER_STATUS);
    } else {
      *result_.mutable_fill_list() = ParseWsFillInfoListV2(document);
      result_.set_type(MsgType::ORDER_FILL);
    }
  }
}

void HuobiParser::ParseRestAccounts() {
  /*
   {
     "data": [
       {
         "id": 100001,
         "type": "spot",
         "subtype": "",
         "state": "working"
       }
       {
         "id": 100002,
         "type": "margin",
         "subtype": "btcusdt",
         "state": "working"
       },
       {
         "id": 100003,
         "type": "otc",
         "subtype": "",
         "state": "working"
       }
     ]
   }
   */
  JsonDocument doc;
  if (!PrepareJsonDocument(&doc)) {
    LOG(ERROR) << "[ParseRestAccounts] Parsing Error." << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  std::map<std::string, std::string> type_id_map;
  if (doc.IsObject() && doc.HasMember("data") && doc["data"].IsArray()) {
    for (const auto& info : doc["data"].GetArray()) {
      if (info.HasMember("type") && info.HasMember("id")) {
        type_id_map.insert_or_assign(info["type"].GetString(), info["id"].GetString());
      }
    }
  }
  CHECK(type_id_map.count("spot") != 0) << JsonToString(doc);
  mb_->PublishEvent("huobi/account_type_id", &type_id_map);
  result_.set_type(MsgType::IGNORABLE);
}

void HuobiParser::ParseRestTransfers(const std::string& id) {
  /*
  {
    "code": 200,
    "message": "success",
    "data": [
    {
      "accountId": 5418476,
      "currency": "usdt",
      "transactAmt": 100.0,
      "transactType": "transfer",
      "transferType": "master-transfer-in",
      "transactId": 2748233033,
      "transactTime": 1663814270371,
      "transferer": 7459131,
      "transferee": 5418476
    },
    {
      "accountId": 5418476,
      "currency": "usdt",
      "transactAmt": -100.0,
      "transactType": "transfer",
      "transferType": "master-transfer-out",
      "transactId": 2748233027,
      "transactTime": 1663814264865,
      "transferer": 5418476,
      "transferee": 7459131
    }
    ]
    "ok": true
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestTransfers] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestTransfers] Invalid transfer format" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    static std::vector<std::string> field =
        {"currency", "transactId", "transactAmt", "transactTime", "transferType"};
    if (!JsonFieldCheck(data, field)) {
      LOG(ERROR) << "[ParseRestTransfers] Invalid transfer format" << JsonToString(data);
      continue;
    }

    std::string record_id = data["transactId"].GetString();
    if (!transfer_first_record_) {
      if (transfer_subaccount_map_[record_id] == "done") {
        continue;
      }
    }
    transfer_subaccount_map_.insert_or_assign(record_id, "done");

    std::string transferType = data["transferType"].GetString();
    auto proto_type = TransferRequestProto::WITHDRAW_SUBACCOUNT;
    if (transferType == "master-transfer-in") {
      proto_type = TransferRequestProto::WITHDRAW_SUBACCOUNT;
    } else if (transferType == "master-transfer-out") {
      proto_type = TransferRequestProto::DEPOSIT_SUBACCOUNT;
    } else {
      LOG(INFO) << "[ParseRestTransfers] ignore transfer info" << JsonToString(data);
      continue;
    }

    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    int64_t create_time = stoll(data["transactTime"].GetString()) * 1e6;
    auto currency = HuobiCurrency::FromNativeStrImpl(data["currency"].GetString(), create_time);
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(proto_type);
    currency_transfer->set_external_transfer_id(record_id);
    currency_transfer->set_qty(std::abs(stod(data["transactAmt"].GetString())));
    currency_transfer->set_create_timestamp(create_time);
    currency_transfer->set_from_address(data["transferee"].GetString());
    currency_transfer->set_to_address(data["transferer"].GetString());
    currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!transfer_subaccount_initialized_set_.count(id)) {
      transfer_subaccount_initialized_set_.insert(id);
      LOG(INFO) << "[ParseRestTransfer] Transfer deposit subaccount ignored: "
                << transfer_subaccount_map_.size();
      return;
    }
  }
  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestTransfers] Transfers parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

auto HuobiParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg.type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::REST_ACCOUNTS:
      ParseRestAccounts();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      ParseBatchCancelResponse();
      break;
    case OrderSubTopicId::MULTI_SUBMIT_RESPONSE:
      ParseBatchSubmitResponse();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestOrdersInfo();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestSpotAccountBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_CROSS_BALANCE:
      ParseRestMarginCrossedAccountBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_NON_CROSS_BALANCE:
      ParseRestMarginIsolatedAccountBalance();
      break;
    case HuobiOrderSubTopicId::REST_DEPOSIT:
      ParseRestDeposit();
      break;
    case HuobiOrderSubTopicId::REST_WITHDRAWAL:
      ParseRestWithdrawal();
      break;
    case OrderSubTopicId::TRANSFER:
      ParseRestTransfers(msg.http_context->extra_info);
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  return result_;
}

}  // namespace coin2::exchange::huobi::order_v1
